package com.guguguweb.rpc.net.client.imp;

import com.guguguweb.rpc.entity.RpcRequest;
import com.guguguweb.rpc.entity.RpcResponse;
import com.guguguweb.rpc.enumeration.ResponseCode;
import com.guguguweb.rpc.enumeration.RpcError;
import com.guguguweb.rpc.exception.RpcException;
import com.guguguweb.rpc.loadbalancer.LoadBalancer;
import com.guguguweb.rpc.loadbalancer.imp.RandomLoadBalancer;
import com.guguguweb.rpc.net.client.GuRpcClient;
import com.guguguweb.rpc.net.client.proxy.ChannelProvider;
import com.guguguweb.rpc.register.ServiceDiscovery;
import com.guguguweb.rpc.register.imp.NacosServiceDiscovery;
import com.guguguweb.rpc.serializer.CommonSerializer;
import com.guguguweb.rpc.util.NacosUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

public class NettyClient implements GuRpcClient {

    private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);


    private EventLoopGroup group;


    private Bootstrap bootstrap;

    private long maxWaitTime ;


    private  ServiceDiscovery serviceDiscovery;

    private  CommonSerializer serializer;


    private UnprocessedRequests unprocessedRequests;


    private ChannelProvider channelProvider;

    public NettyClient(){
        this(1,new RandomLoadBalancer(),DEFAULT_MAX_WAIT_TIME,"localhost",8848);
    }


    public NettyClient(Integer serializerCode, LoadBalancer loadBalancer,long maxWaitTime,String nacosHost,int nacosPort){
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class);

        NacosUtil nacosUtil = new NacosUtil(nacosHost+":"+nacosPort);
        serviceDiscovery = new NacosServiceDiscovery(loadBalancer,nacosUtil);
        serializer = CommonSerializer.getByCode(serializerCode);
        unprocessedRequests = new UnprocessedRequests();
        channelProvider = new ChannelProvider(unprocessedRequests);
        this.maxWaitTime = maxWaitTime;

        logger.info("guRpc 客户端启动成功！");
    }


    @Override
    public ServiceDiscovery getServiceDiscovery() {
        return serviceDiscovery;
    }

    @Override
    public CompletableFuture<RpcResponse> sendRequest(RpcRequest rpcRequest) {
        if (serializer == null) {
            logger.error("未设置序列化器");
            throw new RpcException(RpcError.SERIALIZER_NOT_FOUND);
        }

        CompletableFuture<RpcResponse> resultFuture = new CompletableFuture<>();
        InetSocketAddress address = null;
        try {
            //通过均衡负载，和接口名字获得一个可以访问的 生产者
            address = serviceDiscovery.lookupService(rpcRequest.getInterfaceName());
            Channel channel = channelProvider.get(address,serializer);

            //将新请求放入未处理完的请求中
            unprocessedRequests.put(rpcRequest.getRequestId(),resultFuture);

            channel.writeAndFlush(rpcRequest).addListener((ChannelFutureListener) future -> {
                if(future.isSuccess()){
                    logger.debug(String.format("客户端发送消息：%s", rpcRequest));
                }else {
                    resultFuture.completeExceptionally(future.cause());
                    logger.error("发送消息时有错误发生: ", future.cause());
                    unprocessedRequests.remove(rpcRequest.getRequestId());
                    future.channel().close();
                }
            });
        }catch (Exception e){
            unprocessedRequests.remove(rpcRequest.getRequestId());
            //消费失败，将这个消费者从缓存中删除
            serviceDiscovery.deleteConsumer(rpcRequest.getInterfaceName(),address);

            logger.error(e.getMessage(), e);
            //interrupt()这里作用是给受阻塞的当前线程发出一个中断信号，让当前线程退出阻塞状态，好继续执行然后结束
            Thread.currentThread().interrupt();
            resultFuture.complete(null);
            e.printStackTrace();
        }
        return resultFuture;
    }

    @Override
    public UnprocessedRequests getRequestsMap() {
        return unprocessedRequests;
    }


    @Override
    public long getMaxWaitTime() {
        return maxWaitTime;
    }
}
